package net.bwie.realtime.jtp.dws.log.function;

import net.bwie.realtime.jtp.dws.log.bean.PageViewBean;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewReportWindowFunctionOptimize implements WindowFunction<PageViewBean, String, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow timeWindow,
                      Iterable<PageViewBean> iterable, Collector<String> collector) throws Exception {

        /**
         * s表示分组字段；timeWindow表示时间窗口，可以获取窗口开始时间和结束时间；iterable表示迭代器，表示窗口中所有的数据
         */
        //1-窗口时间
        String windowStart = DateTimeUtil.convertLongToString(timeWindow.getStart(), "yyyy-MM-dd HH:mm:ss");
        String windowEnd = DateTimeUtil.convertLongToString(timeWindow.getEnd(), "yyyy-MM-dd HH:mm:ss");

        //2-获取窗口中计算结果
        PageViewBean bean = iterable.iterator().next();
        Long pvCount = bean.getPvCount();
        Long pvDuringTime = bean.getPvDuringTime();
        Long uvCount = bean.getUvCount();
        Long sessionCount = bean.getSessionCount();

        //3-输出
        String output = windowStart + "," + windowEnd + "," + s + "," +
                pvCount + "," + pvDuringTime + "," + uvCount + "," + sessionCount + "," +
                System.currentTimeMillis();
        collector.collect(output);
    }
}
